package cn.doitedu.day03.demo

import java.sql.{DriverManager, PreparedStatement}

import cn.doitedu.day01.utils.SparkUtil
import cn.doitedu.day03.beans.Student

/**
 * @Date 22.3.31
 * @Created by HANGGE
 * @Description
 */
object D01_数据入库_Mysql {

  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSc

    val rdd1 = sc.textFile("data/students.csv")
    // 待写入到MySQL中的数据信息
    val rdd2 = rdd1.map(line => {
      val arr = line.split(",")
      Student(arr(0).toInt, arr(1), arr(2).toInt, arr(3), arr(4).toDouble, arr(5))
    })
  /*  println(rdd1.getNumPartitions)
    println(rdd2.getNumPartitions)*/
    //========================================获取连接===========================

    // 遍历每一行  Task not serializable
    // 函数运行在不用的机器上
    /**
     * 将每条数据写入到MySQL
     *   1)每天数据执行一次计算逻辑   每条数据获取一次连接  [优化]
     *   2)函数真正的实行是在Executor端的Task线程中  引用的所有的外部变量都要序列化
     *   3)一定触发行动算子 才会真正的执行数据处理任务
     */
    rdd2.foreach(stu=>{
      val conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/doit30?characterEncoding=UTF-8", "root", "root")
      val ps: PreparedStatement = conn.prepareStatement("insert  into  tb_student values (?,?,?,?,?,?)")
      // 每个学生写到mysql中
      //1) 预编译  sql
      ps.setInt(1, stu.sid)
      ps.setString(2,stu.name)
      ps.setInt(3 , stu.age)
      ps.setString(4,stu.gender)
      ps.setDouble(5,stu.score)
      ps.setString(6,stu.grade)
      // 2) 执行SQL语句
      ps.executeUpdate()
    })


   // conn.close()
    sc.stop()

  }

}
